package com.shujia.spark.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo11Action {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("action")
      .setMaster("local")

    val sc = new SparkContext(conf)

    /**
      * action算子，触发任务执行，每一个action算子都会触发一个job
      */

    val students: RDD[String] = sc.textFile("data/students.txt")

    /**
      * count ：统计RDD的数量
      */

    val stuCount: Long = students.count()

    println(stuCount)

    /**
      * collect 将rdd的数据拉取到内存中，变成一个数组
      * 如果rdd的数据量超过了内存限制会出现内存溢出
      */

    val studentArray: Array[String] = students.collect()

    /**
      * reduce 全局聚合
      */

    val str: String = students.reduce(_+_)

    /**
      * foreach: 遍历rdd中的数据
      * foreachPartition: 遍历一个分区, 一般用在将rdd的数据保存到外部数据库的时候
      */

    students.foreach(println)

    students.foreachPartition((s: Iterator[String]) => {
      s.foreach(println)
    })

    /**
      * saveAsTextFile保存数据到hdfs,   如果输出目录已存在会报错
      *
      */

    val configuration = new Configuration()

    val fileSystem: FileSystem = FileSystem.get(configuration)

    if (fileSystem.exists(new Path("data/out1"))){
      //递归删除目录
      fileSystem.delete(new Path("data/out1"),true)
    }

    students.saveAsTextFile("data/out1")

  }

}
